package drop

import cn.itcast.dmp.config.AppConfigHelper
import org.apache.spark.sql.{DataFrame, SparkSession}

object SparkDatasetDuplicate {
  def main(args: Array[String]): Unit = {

    // 1、获取SparkSession实例对象
    val spark = SparkSession.builder()
      .master("local[4]")
      .appName(this.getClass.getSimpleName.stripSuffix("$"))
      .config("spark.sql.shuffle.partitions", "4")
      .getOrCreate()
    import spark.implicits._
    // 2、读取Kudu表的数据
    import cn.itcast.dmp.untils.KuduUtils._
    val odsDF = spark.readKuduTable(AppConfigHelper.AD_MAIN_TABLE_NAME).get
    // TODO: 获取 经度、维度及GeoHash值
    val geoDF: DataFrame = odsDF.select('longitude, 'latitude, $"geoHash")
    println(s"读取Kudu表中数据条目数：${geoDF.count()}")
    geoDF.show(10, truncate = false)
    // 调用 distinct 函数去重
    val disDF = geoDF.distinct()
    println(s"去重后:${disDF.count()}")
    disDF.show(10, truncate = false)
    println("=============================================")
    val delDF = geoDF.drop($"longitude")
    delDF.printSchema()
    delDF.show(10, truncate = false)
    // 当应用运行完成，关闭资源
    spark.stop()
  }
}